package Demo1;
import java.io.IOException;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Map;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

public class DemoTopology {
	
	public static void main(String[] args) throws Exception {

	    TopologyBuilder builder = new TopologyBuilder();

	    builder.setSpout("spout", new DemoSpout(), 1);

	    builder.setBolt("add1", new DemoBlotAdd(1), 1).shuffleGrouping("spout");
	    builder.setBolt("add2", new DemoBlotAdd(2), 1).shuffleGrouping("spout");
	    builder.setBolt("add3", new DemoBlotAdd(3), 1).shuffleGrouping("spout");
	    builder.setBolt("add4", new DemoBlotAdd(4), 1).shuffleGrouping("spout");
	    builder.setBolt("add5", new DemoBlotAdd(5), 1).shuffleGrouping("spout");
	    builder.setBolt("printBolt", new DemoBlotPrint()).shuffleGrouping("add1").shuffleGrouping("add2").shuffleGrouping("add1").shuffleGrouping("add3").shuffleGrouping("add4").shuffleGrouping("add5");
	    
	    Config conf = new Config();
	    conf.setDebug(true);


	    if (args != null && args.length > 0) {
	      conf.setNumWorkers(3);

	      StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
	    }
	    else {
	      conf.setMaxTaskParallelism(1);

	      LocalCluster cluster = new LocalCluster();
	      cluster.submitTopology("word-count", conf, builder.createTopology());
	    }
	  }
}
